package com.lry;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.LongAdder;

@Slf4j
public class Node {

    private String name;
    private NodeRole nodeRole;
    private List<Node> nodes;
    private Node master;
    private long howMsNotReceiveHeartbeat;
    private Random random = new Random();
    private volatile long lastReceiveHeartbeatTime;
    private LongAdder term;//任期编号
    private LongAdder vote;//收到了多少票

    public Node(String name){
        this.nodes = new ArrayList<>();
        this.name = name;
        this.nodeRole = NodeRole.FOLLOWER;
        //随机超时时间设置为100-300ms
        this.howMsNotReceiveHeartbeat = random.nextInt(201)+100;
        this.term = new LongAdder();
        this.vote = new LongAdder();
    }

    public void registerNodes(List<Node> inNodes,Node exceptNode){
        for (Node node : inNodes) {
            if(null==node||node==exceptNode){
                continue;
            }
            nodes.add(node);
        }
        lastReceiveHeartbeatTime = System.currentTimeMillis();
        this.heartbeatCheck();
    }

    public void candidate(){
        this.vote.reset();//每次选举前需要把上次选举的投票清空

        this.nodeRole = NodeRole.CANDIDATE;
        this.term.increment();//每轮选举算法自增自己的任期编号
        this.vote.increment();//给自己投一票

        //发送自己的选票信息给所有节点
        for (Node node : nodes) {
            //拉票
            boolean reqVoteResult = node.reqVote(this.name,this.term.intValue());
            if(reqVoteResult){
                this.vote.increment();
            }
        }

        //统计票数过不过半
        if(this.vote.intValue()>(this.nodes.size()+1)/2){
            log.info("节点={}成为leader，vote={}，开始周期性发送心跳给所有节点",name,this.vote.intValue());
            //成为leader
            this.nodeRole = NodeRole.LEADER;
            sendHeartbeat();
        }
    }

    /**
     * 拉票函数
     * @param nodeName 请求拉票的节点名字
     * @param termValue 请求拉票的节点term值
     * @return
     */
    private boolean reqVote(String nodeName, int termValue) {
        log.info("节点{}收到节点{}的拉票信息，他的term={}，我的term={}",this.name,nodeName,termValue,this.term.intValue());
        if(termValue>this.term.intValue()){
            return true;
        }
        log.info("节点{}的选票已投出去",this.name);
        return false;
    }

    private void sendHeartbeat(){
        CompletableFuture.runAsync(()->{
            while(true){
                //开始周期性的发送心跳给所有节点
                for (Node node : nodes) {
                    CompletableFuture.runAsync(()->{
                        node.heartbeat(name);
                    });
                }
                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    public void heartbeatCheck(){
        CompletableFuture.runAsync(()->{
           while(true){
               long cur = System.currentTimeMillis();
               //超时了，开始去参与选举算法
               if(cur-lastReceiveHeartbeatTime>howMsNotReceiveHeartbeat&&this.nodeRole!=NodeRole.LEADER){
                   candidate();
               }

               try {
                   Thread.sleep(10);
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }

           }
        });
    }

    //接收master的心跳
    public void heartbeat(String nodeName){
        log.info("节点{}收到节点={}的心跳",this.name,nodeName);
        lastReceiveHeartbeatTime = System.currentTimeMillis();
    }


}
